package com.fanjj.spark.test

import java.util
import java.util.{ArrayList, Arrays, List, Random, UUID}

import com.fanjj.spark.util.{DateUtils, StringUtils}
import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, Row, RowFactory, SQLContext}
import org.apache.spark.sql.types.{DataTypes, StructType}
;


object MockData {
  /**
    * 模拟数据
    *
    * @param sc
    * @param sqlContext
    */
  def mock(sc: SparkContext, sqlContext: SQLContext): Unit = {
    val rows = new util.ArrayList[Row]
    val searchKeywords = Array[String]("火锅", "蛋糕", "重庆辣子鸡", "重庆小面", "呷哺呷哺", "新辣道鱼火锅", "国贸大厦", "太古商场", "日本料理", "温泉")
    val date = DateUtils.getTodayDate
    val actions = Array[String]("search", "click", "order", "pay")
    val random = new Random
    val i = 0

    for (i <- 0 until 100) {
      val userid = random.nextInt(100)

      for (j <- 0 until 10) {
        val sessionid = UUID.randomUUID.toString.replace("-", "")
        val baseActionTime = date + " " + random.nextInt(23)
        var clickCategoryId: Long = 0
        var k = 0
        for (k <- 1 until random.nextInt(100)) {
          val pageid = random.nextInt(10)
          val actionTime = baseActionTime + ":" + StringUtils.fulfuill(String.valueOf(random.nextInt(59))) + ":" + StringUtils.fulfuill(String.valueOf(random.nextInt(59)))
          var searchKeyword: String = null
          var clickProductId: Long = 0
          var orderCategoryIds: String = null
          var orderProductIds: String = null
          var payCategoryIds: String = null
          var payProductIds: String = null
          val action = actions(random.nextInt(4))
          if ("search" == action) searchKeyword = searchKeywords(random.nextInt(10))
          else if ("click" == action) {
            if (clickCategoryId == null) clickCategoryId = random.nextInt(100).toLong
            clickProductId = random.nextInt(100).toLong
          }
          else if ("order" == action) {
            orderCategoryIds = random.nextInt(100).toString
            orderProductIds = String.valueOf(random.nextInt(100))
          }
          else if ("pay" == action) {
            payCategoryIds = String.valueOf(random.nextInt(100))
            payProductIds = String.valueOf(random.nextInt(100))
          }
          val row = Row(date, userid, sessionid, pageid, actionTime, searchKeyword, clickCategoryId, clickProductId, orderCategoryIds, orderProductIds, payCategoryIds, payProductIds, random.nextInt(10).toLong)
          rows.add(row)
        }
      }


      val schema = DataTypes.createStructType(
        util.Arrays.asList(DataTypes.createStructField("date", DataTypes.StringType, true),
          DataTypes.createStructField("user_id", DataTypes.LongType, true),
          DataTypes.createStructField("session_id", DataTypes.StringType, true),
          DataTypes.createStructField("page_id", DataTypes.LongType, true),
          DataTypes.createStructField("action_time", DataTypes.StringType, true),
          DataTypes.createStructField("search_keyword", DataTypes.StringType, true),
          DataTypes.createStructField("click_category_id", DataTypes.LongType, true),
          DataTypes.createStructField("click_product_id", DataTypes.LongType, true),
          DataTypes.createStructField("order_category_ids", DataTypes.StringType, true),
          DataTypes.createStructField("order_product_ids", DataTypes.StringType, true),
          DataTypes.createStructField("pay_category_ids", DataTypes.StringType, true),
          DataTypes.createStructField("pay_product_ids", DataTypes.StringType, true),
          DataTypes.createStructField("city_id", DataTypes.LongType, true)))
      val df: DataFrame = sqlContext.createDataFrame(rows, schema)
      df.createOrReplaceTempView("user_visit_action")
      println(df.show())

      /**
        * ==================================================================
        **/
      rows.clear()
      val sexes = Array[String]("male", "female")

      for (i <- 0 until 100) {
        val userid = i
        val username = "user" + i
        val name = "name" + i
        val age = random.nextInt(60)
        val professional = "professional" + random.nextInt(100)
        val city = "city" + random.nextInt(100)
        val sex = sexes(random.nextInt(2))
        val row = Row(userid, username, name, age, professional, city, sex)
        rows.add(row)


      }

      val schema2 = DataTypes.createStructType(
        util.Arrays.asList(
          DataTypes.createStructField("user_id", DataTypes.LongType, true),
          DataTypes.createStructField("username", DataTypes.StringType, true),
          DataTypes.createStructField("name", DataTypes.StringType, true),
          DataTypes.createStructField("age", DataTypes.IntegerType, true),
          DataTypes.createStructField("professional", DataTypes.StringType, true),
          DataTypes.createStructField("city", DataTypes.StringType, true),
          DataTypes.createStructField("sex", DataTypes.StringType, true)))
      val df2: DataFrame = sqlContext.createDataFrame(rows, schema2);


      df2.createOrReplaceTempView("user_info");
      println(df2.show())
      rows.clear()
      val productStatus = Array[Int](0, 1)

      for (i <- 0 until 100) {
        val productId = i
        val productName = "product" + i
        val extendInfo = "{\"product_status\": " + productStatus(random.nextInt(2)) + "}"
        val row = Row(productId, productName, extendInfo)
        rows.add(row)

      }

      val schema3 = DataTypes.createStructType(
        util.Arrays.asList(
          DataTypes.createStructField("product_id", DataTypes.LongType, true),
          DataTypes.createStructField("product_name", DataTypes.StringType, true),
          DataTypes.createStructField("extend_info", DataTypes.StringType, true)))
      val df3: DataFrame = sqlContext.createDataFrame(rows, schema3);
      println(df3.show())

      df3.createOrReplaceTempView("product_info");
    }
  }
}
